package org.example.common.configs.queue.disruptor;

import com.alibaba.fastjson.JSON;
import com.lmax.disruptor.RingBuffer;
import com.lmax.disruptor.dsl.Disruptor;
import lombok.AllArgsConstructor;
import lombok.extern.slf4j.Slf4j;
import org.springframework.beans.factory.DisposableBean;

import java.io.Serializable;

/**
 * @author weigang
 * @create 2022-07-11 14:28
 */
@AllArgsConstructor
@Slf4j
public class DisruptorEventProducerWorker implements DisposableBean {
    private RingBuffer<DisruptorEvent> ringBuffer;
    private Disruptor disruptor;
    private DisruptorStat disruptorStat;

    public void publishEvent(Serializable event) {
        // ringBuffer是个队列，其next方法返回的是下最后一条记录之后的位置，这是个可用位置
        // 获取到 sequence 之后，必须保证这个sequence被publish，这里也会唤醒消费者
        // 若是沒有足够的空间，那么 next() 方法就会被阻塞，LockSupport#parkNanos
        // todo hhr 借助tryNext方法与Guava Retrying实现带超时的重试机制
        /* 发布事件
        * Disruptor的事件Event的发布Publish过程是一个两阶段提交过程:
            第一步: 先从RingBuffer获取下一个可以写入事件的序号
            第二步: 获取对应的事件Event对象,将数据写入事件对象
            第三步: 将事件提交到RingBuffer
        * Disruptor中要求RingBuffer.publish()方法必须要被调用.也就是说,即使发生异常,也要执行publish()方法,
        * 这就要求调用者Producer在事件处理的实现上要判断携带的数据的正确性和完整性
        * */
        // 第一步
        long sequence = ringBuffer.next();
        try {
            // 第二步
            DisruptorEvent disruptorEvent = ringBuffer.get(sequence);
            disruptorEvent.setEvent(event);
            log.info("Producder Event Success ,content ：{}, Thread-id :{}", JSON.toJSON(event),Thread.currentThread().getId());
            // 当前正在执行或者待执行的事件数量 +1
            disruptorStat.currentEventCounter.incrementAndGet();
        } finally {
            // 第三步
            ringBuffer.publish(sequence);
        }
    }

    @Override
    public void destroy() throws Exception {
        disruptor.shutdown();
    }

    public void start() {
        disruptor.start();
    }

}
